package com.example.demo.nios.selecor;

import io.netty.channel.ServerChannel;
import io.swagger.models.auth.In;
import lombok.Data;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

// 由于是非阻塞的，所以可以引入线程池，阻塞式的不能采用线程池

public class SocketChannelServer {

    public static volatile Map<Integer,SocketChannel> keys = Collections.synchronizedMap(new HashMap<Integer, SocketChannel>());//考虑并发 移除也可以用ConcurrentHashMap
    public static volatile Map<Integer,ServerSocketChannel> SERVER_MAP = Collections.synchronizedMap(new HashMap<Integer,ServerSocketChannel>());
    static ConcurrentLinkedQueue<MsgWrapper> msgQueue = new ConcurrentLinkedQueue<MsgWrapper>();
    static Charset charset = Charset.forName("utf-8");
    public static volatile Map<Integer, Selector> selector_map = Collections.synchronizedMap(new HashMap<>());


    public static void main(String[] args) {
       AcceptSocketThread.open(1024);
    }
    public static void add(){

    }
    @Data
    public static class AcceptSocketThread {


        volatile boolean runningFlag = true;

        public static void open(Integer port) {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            boolean runningFlag = true;
            try {
                ServerSocketChannel serverChannel = ServerSocketChannel.open();
                ServerSocket socket = serverChannel.socket();
                socket.bind(new InetSocketAddress(port));
                serverChannel.configureBlocking(false);
                SERVER_MAP.put(port, serverChannel);
                Selector selector = Selector.open();
                serverChannel.register(selector, SelectionKey.OP_ACCEPT);

                while (runningFlag) {
                    int n = selector.select();
                    //没有就绪的通道则什么也不做
                    if (n == 0) {
                        continue;
                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey sk = iterator.next();

                        if (sk.isAcceptable()) {
                            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) sk.channel();
                            SocketChannel channel = serverSocketChannel.accept();
                            channel.configureBlocking(false);
                            channel.register(selector,SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                            System.out.println("一个客户端上线，占用端口 ：" + channel.socket().getPort());
                        } else if (sk.isReadable()) {
                            SocketChannel channel = (SocketChannel) sk.channel();
                            StringBuffer stb = new StringBuffer();
                            while (channel.read(buffer) > 0) {
                                buffer.flip();
                                stb.append(charset.decode(buffer).toString());
                                buffer.clear();
                            }
                            MsgWrapper msgWrapper = new MsgWrapper();
                            msgWrapper.key = channel.socket().getPort();
                            msgWrapper.msg = stb.toString();
                            msgQueue.add(msgWrapper);
                        } else if (sk.isWritable()) {
                            SocketChannel channel = (SocketChannel) sk.channel();
                            if (!msgQueue.isEmpty()) {
                                MsgWrapper poll = msgQueue.poll();
                                channel.write(charset.encode(poll.msg));
                            }
                        } else if (!sk.isConnectable()) {
                            sk.cancel();
                            selector.close();
                            runningFlag = false;
                        } else if (sk.isValid()) {
                            sk.cancel();
                            selector.close();
                            runningFlag = false;
                        }

                        iterator.remove();


                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }

    /**
     * 由于读是非阻塞式的，所以没必要一个socketChannel一个线程
     * 也可以通过线程池来执行，此处只做实例，学习，不扩展
     * @author liangpro
     */
   public static class ReadMsgThread extends Thread{
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        public void run(){
            try {
                int num = 0;
                for (;;) {
                    Iterator<Integer> ite = keys.keySet().iterator();
                    while (ite.hasNext()) {
                        int key = ite.next();
                        StringBuffer stb = new StringBuffer();
                        try{
                            while((num = keys.get(key).read(buffer)) > 0 ){
                                buffer.flip();
                                stb.append(charset.decode(buffer).toString());
                                buffer.clear();
                            }
                            if(stb.length() > 0){
                                MsgWrapper msg = new MsgWrapper();
                                msg.key = key;
                                msg.msg = stb.toString();
                                System.out.println("端口：" + msg.key + "的通道,读取到的数据" + msg.msg);
                                String[] split = msg.msg.split("&&");
                                if("123".equals(split[0]) && "456".equals(split[1])){
                                    msg.msg += "成功";
                                }
                                else{
                                    msg.msg += "失败";
                                }
                                msgQueue.add(msg);
                            }
                        }catch(Exception e){
                            System.out.println("error: 端口占用为：" + keys.get(key).socket().getPort() + ",的连接的客户端下线了");
                            ite.remove();
                            e.printStackTrace();//这只是个输出语句
                        }
                    }
                    sleep(1000);
//                    System.out.println("读取线程监听中......");
                }
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }



    }

   public static class ConsumerMsgThread extends Thread{
        public volatile boolean isRunningFlag = false;

        public void run (){
            isRunningFlag = true;
            try {
//				sleep(10000);
                while (isRunningFlag) {
                    MsgWrapper msg = msgQueue.poll();
                    for (; null != msg; msg = msgQueue.poll()) {
                        SocketChannel channel = keys.get(msg.key);
                        channel.write(charset.encode("response:" + msg.msg));
                    }
                    sleep(1000);
//                    System.out.println("响应线程响应中......");
                }
            } catch (IOException | InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

    static class MsgWrapper {
        public int key;
        public String msg;
    }
}
